package org.spark.core.action.java;

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.spark.sparkUtil.SparkJavaContextUtil;

import java.util.Arrays;
import java.util.List;

/**
 * 功能概述:
 * Datetime:    2020/5/28   19:18
 * Author:   某人的目光
 */
public class Collect {
    public static void main(String[] args) {
        JavaSparkContext sc = SparkJavaContextUtil.getSparkStart("Collect");
        // 有一个集合，里面有1到10,10个数字，现在要对10个数字进行累加
        List<Integer> numberList = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
        JavaRDD<Integer> numbers = sc.parallelize(numberList);

        // 使用map操作将集合中所有数字乘以2
        JavaRDD<Integer> doubleNumbers = numbers.map(

                new Function<Integer, Integer>() {

                    private static final long serialVersionUID = 1L;

                    public Integer call(Integer v1) throws Exception {
                        return v1 * 2;
                    }

                });

        // 不用foreach action操作，在远程集群上遍历rdd中的元素
        // 而使用collect操作，将分布在远程集群上的doubleNumbers RDD的数据拉取到本地
        // 这种方式，一般不建议使用，因为如果rdd中的数据量比较大的话，比如超过1万条
        // 那么性能会比较差，因为要从远程走大量的网络传输，将数据获取到本地
        // 此外，除了性能差，还可能在rdd中数据量特别大的情况下，发生oom异常，内存溢出
        // 因此，通常，还是推荐使用foreach action操作，来对最终的rdd元素进行处理
        List<Integer> doubleNumberList = doubleNumbers.collect();
        for(Integer num : doubleNumberList) {
            System.out.println(num);
        }

        // 关闭JavaSparkContext
        sc.close();
    }
}
